package main

import (
	"fmt"
	"net"
	"strings"
	"sync"
	"time"
)

const (
	tcpBufSize int = 4096

	tcpPackageStartTag     uint16 = 0x1F1F
	tcpPackageprotoVersion uint16 = 0x0001

	LOGIN_MEDIA_SERVER_CMD                      uint16 = 100
	LOGIN_MEDIA_SERVER_CMD_REP                  uint16 = 102
	LOGOUT_MEDIA_SERVER_CMD                     uint16 = 103
	REQUEST_PUSH_STREAM_TO_MEDIA_SERVER_CMD     uint16 = 105
	REQUEST_PUSH_STREAM_TO_MEDIA_SERVER_CMD_REP uint16 = 106
	CANCEL_PUSH_STREAM_TO_MEDIA_SERVER_CMD      uint16 = 107
	CANCEL_PUSH_STREAM_TO_MEDIA_SERVER_CMD_REP  uint16 = 108
	REQUEST_STREAM_FROM_MEDIA_SERVER_CMD        uint16 = 109
	REQUEST_STREAM_FROM_MEDIA_SERVER_CMD_REP    uint16 = 110
	CANCEL_STREAM_TO_MEDIA_SERVER_CMD           uint16 = 111
	CANCEL_STREAM_TO_MEDIA_SERVER_CMD_REP       uint16 = 112

	SERVER_CLIENT_KEEPALIVE_CMD     uint16 = 1000
	SERVER_CLIENT_KEEPALIVE_CMD_REP uint16 = 1001

	maxTcpClientNum int = 20

	tcpClientHeartbeatCheckInternal int64 = 5
)

var (
	tcpClientListRwLock sync.RWMutex
	tcpClientList       = make([]TcpClientAddrType, maxTcpClientNum)

	tcpClientHeartbeatOverTime int64 = tcpClientHeartbeatCheckInternal * 4

	tcpChans [maxTcpClientNum]chan []byte
)

func tcpServer(listenPort int) int {
	initTcp()
	listenAddr := fmt.Sprintf("0.0.0.0:%d", listenPort)
	listener, err := net.Listen("tcp", listenAddr)
	if err != nil {
		fmt.Println("listen fail ! port =", listenPort)
		return -1
	}

	fmt.Println("tcp server listening at", listenAddr)
	go heartbeatCheckThread()

	for {
		conn, err := listener.Accept()
		if err != nil {
			fmt.Println("listener.Accept fail !")
			time.Sleep(2 * time.Second)
		}

		go tcpRecvOneClientDataThread(conn)
	}
}

func initTcp() {
	for i := range tcpChans {
		tcpChans[i] = make(chan []byte, tcpBufSize*2)
	}
}

func heartbeatTimeoutDeal(tcpAddrString string) {
	noticeClientOutLineByTcpAddrString(tcpAddrString)
}

func heartbeatCheckThread() {
	var minTime int64
	for {
		minTime = time.Now().Unix() - tcpClientHeartbeatOverTime

		tcpClientListRwLock.RLock()
		for i := range tcpClientList {
			e := &tcpClientList[i]
			if e.busy == false {
				continue
			} else {
				if e.lastHeartbeatTime < minTime {
					tcpAddrString := e.conn.RemoteAddr().String()
					fmt.Println("client", i, "heartbeat over time !", tcpAddrString)
					heartbeatTimeoutDeal(tcpAddrString)
					e.conn.Close()
					e.busy = false
				}
			}
		}
		tcpClientListRwLock.RUnlock()

		time.Sleep(time.Duration(tcpClientHeartbeatCheckInternal) * time.Second)
	}
}

func tcpRecvOneClientDataThread(conn net.Conn) {
	dataBufsIdx := -1

	tcpClientListRwLock.Lock()
	for i := range tcpClientList {
		e := &tcpClientList[i]
		if e.busy == false {
			e.busy = true
			e.conn = conn
			e.lastHeartbeatTime = time.Now().Unix()
			dataBufsIdx = i
			break
		}
	}
	tcpClientListRwLock.Unlock()

	if dataBufsIdx < 0 {
		fmt.Println("no space to add new tcp client !", conn.RemoteAddr().String())
		conn.Close()
		return
	}

	dataBuf := make([]byte, tcpBufSize)
	tcpChan := make(chan []byte, tcpBufSize*10)
	go tcpDataHandleThread(conn, tcpChan)

	for {
		packageSize, err := conn.Read(dataBuf[0:])
		if err != nil {
			fmt.Println("disconnect from", conn.RemoteAddr().String())
			conn.Close()
			return
		}

		if packageSize <= 0 || packageSize > tcpBufSize {
			fmt.Println("packageSize=", packageSize)
			continue
		}

		tcpChan <- dataBuf[0:packageSize]
	}
}

func tcpDataHandleThread(conn net.Conn, tcpChan <-chan []byte) {
	dataTmp := make([]byte, tcpBufSize*2)
	headIdx := 0
	curIdx := 0
	tmpBuf := make([]byte, tcpBufSize)

	for {
		chanData := <-tcpChan
		chanDataLen := len(chanData)
		copy(dataTmp[curIdx:], chanData)
		curIdx = curIdx + chanDataLen
		dataSize := curIdx - headIdx
		if dataSize > 100 {
			fmt.Println("chanDataLen=", chanDataLen, "org dataSize=", dataSize)
		}

		for {
			if dataSize > 100 {
				fmt.Println(headIdx, curIdx, dataSize)
			}
			b := dataTmp[headIdx : headIdx+dataSize]
			if len(b) <= protoHeadSize {
				fmt.Println("len(b) too small !", len(b), string(b))
				break
			}

			head := &ProtoHeadType{uint16(b[0]) | uint16(b[1])<<8, uint16(b[2]) | uint16(b[3])<<8, uint16(b[4]) | uint16(b[5])<<8, uint16(b[6]) | uint16(b[7])<<8}
			realDataSize := int(head.bodySize)
			packageSize := realDataSize + protoHeadSize

			sizeDiff := dataSize - packageSize
			if sizeDiff < 0 {
				fmt.Println("sizeDiff=", sizeDiff, "dataSize=", dataSize, "packageSize=", packageSize)
				fmt.Println("xml1:", string(dataTmp[headIdx:curIdx]))
				if sizeDiff < 0-tcpBufSize {
					fmt.Println("terrible error ! clear all !")
					idx := strings.Index(string(dataTmp[headIdx:]), xmlHead)
					if idx < 0 {
						fmt.Println("try to fix fail !")
					} else {
						headIdx = headIdx + idx - protoHeadSize
						dataSize = curIdx - headIdx
					}
				} else {
					curIdx = headIdx + dataSize
					fmt.Println("222:curIdx, headIdx, dataSize = ", curIdx, headIdx, dataSize)
				}
				break
			}

			if head.tag != tcpPackageStartTag {
				fmt.Printf("head.tag error ! tag = %x\n", head.tag)
				fmt.Println("xml3:", string(dataTmp[headIdx:]))
				curIdx = headIdx + dataSize
				headIdx = curIdx
				dataSize = dataSize - packageSize
				continue
			}

			//fmt.Println("dataTmp[headIdx+protoHeadSize:headIdx+realDataSize]=", string(dataTmp[headIdx+protoHeadSize:headIdx+protoHeadSize+realDataSize]))
			copy(tmpBuf, dataTmp[headIdx+protoHeadSize:headIdx+protoHeadSize+realDataSize])
			tmpBuf[realDataSize] = 0
			c := tmpBuf[0 : realDataSize+1]
			tcpParsePackage(head, &c, &conn)

			headIdx = headIdx + packageSize
			dataSize = dataSize - packageSize
			if dataSize == 0 {
				break
			}
		}

		if dataSize == 0 {
			headIdx = 0
			curIdx = 0
		} else {
			unParseData := dataTmp[headIdx:curIdx]
			copy(dataTmp, unParseData)
			headIdx = 0
			curIdx = dataSize
			dataTmp[dataSize] = 0
			fmt.Println("dataSize=", dataSize, "unParseData:", string(dataTmp[0:dataSize]))
		}
	}
}

func tcpParsePackage(head *ProtoHeadType, recvXmlData *[]byte, conn *net.Conn) {
	if head.cmd != SERVER_CLIENT_KEEPALIVE_CMD {
		fmt.Println("tcpParsePackage: head=", head)
		//fmt.Println("recvXmlData=", string(*recvXmlData))
	}

	switch head.cmd {
	case LOGIN_MEDIA_SERVER_CMD:
		LOGIN_MEDIA_SERVER_CMD_handle(recvXmlData, conn)
	case LOGOUT_MEDIA_SERVER_CMD:
		LOGOUT_MEDIA_SERVER_CMD_handle(recvXmlData, conn)
	case REQUEST_PUSH_STREAM_TO_MEDIA_SERVER_CMD:
		REQUEST_PUSH_STREAM_TO_MEDIA_SERVER_CMD_handle(recvXmlData, conn)
	case CANCEL_PUSH_STREAM_TO_MEDIA_SERVER_CMD:
		CANCEL_PUSH_STREAM_TO_MEDIA_SERVER_CMD_handle(recvXmlData, conn)
	case REQUEST_STREAM_FROM_MEDIA_SERVER_CMD:
		REQUEST_STREAM_FROM_MEDIA_SERVER_CMD_handle(recvXmlData, conn)
	case CANCEL_STREAM_TO_MEDIA_SERVER_CMD:
		CANCEL_STREAM_TO_MEDIA_SERVER_CMD_handle(recvXmlData, conn)

	case SERVER_CLIENT_KEEPALIVE_CMD:
		SERVER_CLIENT_KEEPALIVE_CMD_handle(recvXmlData, conn)
	default:
		fmt.Println("unknow tcp package cmd ! ", head.cmd)
	}
}
